/*
 * simplemq_mqtt.hpp
 *
 *  Created on: 2021年10月10日
 *      Author: guyadong
 */
#ifdef SIMPLEMQ_MQTT_SIMPLEMQ_REDIS_HPP_
#error simplemq_redis.hpp included,cant include simplemq_mqtt.hpp at the same time
#error (不允许同时 include simplemq_mqtt.hpp和simplemq_redis.hpp)
#endif 
#ifndef SIMPLEMQ_MQTT_SIMPLEMQ_MQTT_HPP_
#define SIMPLEMQ_MQTT_SIMPLEMQ_MQTT_HPP_
#include <map>
#include <mutex>
#include <limits>
#include <simplemq/simplemq.hpp>
#include <winsock_initializer.h>
#include <uri_utils.hpp>
#include <resource_manager.hpp>
#include <mqtt/client.h>
#include <mqtt/async_client.h>
namespace simplemq{
	/**
	 * 根据指定的 host/port/username/password 创建 mqtt uri 实例
	 * @param host host to be connected to
	 * @param port port to be connected to
	 * @param username user name for redis authorization
	 * @param password password for redis authorization
	 * @param timeout_ms
	 * @param max_reconnects
	 * @param reconnect_interval_ms
	 * @return uri instance
	 */
	inline uri make_uri(const std::string &host, size_t port,
		const std::string &username = "",
		const std::string &password = "",
		std::uint32_t timeout_ms = std::numeric_limits<std::uint32_t>::max(),
		std::int32_t max_reconnects = std::numeric_limits<std::uint32_t>::max(),
		std::uint32_t reconnect_interval_ms = std::numeric_limits<std::uint32_t>::max()) {
		return simplemq::make_uri(MQTT_SCHEMA, host, port, username, password, timeout_ms, max_reconnects, reconnect_interval_ms);
	}
	/**
	 * 对uri对象进行归一化，输出标准的mqtt uri(tcp://username:password@host:port),不包含请求参数
	 * @param uri
	 * @return uri instance
	 */
	inline uri normalize(const uri &uri) 
	{
		return make_uri(uri.get_host(), uri.get_port(), uri.get_username(), uri.get_password());
	}
	inline uri host_and_port(const uri &uri)
	{
		return make_uri(uri.get_host(), uri.get_port());
	}
	struct mq_client_params : public base_mq_client_params {
		mq_client_params(
			const std::string &host = DEFAULT_HOST,
			int port = DEFAULT_MQTT_PORT,
			const std::string &password = DEFAULT_PASSWORD,
			const std::string &username = DEFAULT_PASSWORD,
			std::uint32_t timeout_ms = DEFAULT_TIMEOUT_MS,
			std::int32_t max_reconnects = DEFAULT_MAX_RECONNECTS,
			std::uint32_t reconnect_interval_ms = DEFAULT_RECONNECT_INTERVAL_MS)
			: base_mq_client_params(MQTT_SCHEMA,DEFAULT_MQTT_PORT,host, port, password, username, timeout_ms, max_reconnects, reconnect_interval_ms) {}
		mq_client_params(const uri&uri) : base_mq_client_params(MQTT_SCHEMA,DEFAULT_MQTT_PORT,parse_components(uri)) {}
		_BASIC_JSON_TPL_DECLARATION_
		mq_client_params(const _BASIC_JSON_TPL_ &params) : base_mq_client_params(MQTT_SCHEMA,DEFAULT_MQTT_PORT,params) {}
		/** 输出标准的mqtt uri(tcp://username:password@host:port),不包含请求参数 */
		uri normalize_uri()const
		{
			return simplemq::make_uri(schema,host, port, username, password);
		}
	};
	inline std::string make_client_id()
	{
		auto now_mills = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch());
		return MQTT_CLIENT_ID_PREFIX + gdface::string_utils::format("_%lX", (long)now_mills.count());
	}
	using mqtt_client_ptr = std::shared_ptr<::mqtt::client>;
	using mqtt_async_client_ptr = std::shared_ptr<::mqtt::async_client>;
	using dispatch_fun = std::function<void(const std::string & channel, const std::string & message)>;

	/**
	 * 创建 mqtt::client/mqtt::async_client 实例
	 * @param connOpts [out] connect options
	 * @param target host and port to connect
	 * @param params parameters to be connected to
	 * @return
	 */
	template<typename CLI>
	inline std::shared_ptr<CLI> makeClient(::mqtt::connect_options &connOpts, const uri& target, const mq_client_params& params)
	{
		auto id = make_client_id();
		SAMPLE_DEBUG_LOG("create mqtt client id:{}", id);
		auto cli = std::make_shared<CLI>(host_and_port(target).to_string(), id);
		if (!params.username.empty())
		{
			connOpts.set_user_name(params.username);
		}
		if (!params.password.empty())
		{
			connOpts.set_password(params.password);
		}
		if (params.timeout_ms != std::numeric_limits<std::uint32_t>::max())
		{
			connOpts.set_connect_timeout(std::chrono::milliseconds(params.timeout_ms));
		}
		if (params.reconnect_interval_ms != std::numeric_limits<std::uint32_t>::max())
		{
			connOpts.set_automatic_reconnect(
				std::chrono::milliseconds(params.reconnect_interval_ms),
				std::chrono::milliseconds(params.reconnect_interval_ms));
		}
		return cli;
	}
	/**
	 * 创建 mqtt::client 实例
	 * @param connOpts [out] connect options
	 * @param target host and port to connect
	 * @param params parameters to be connected to
	 * @return
	 */
	inline mqtt_client_ptr makeClient(::mqtt::connect_options &connOpts, const uri& target, const mq_client_params& params)
	{
		return makeClient<::mqtt::client>(connOpts,target,params);
	}

	/**
	 * 创建 mqtt::client 实例
	 * @param connOpts [out] connect options
	 * @param params parameters to be connected to
	 * @return
	 */
	inline mqtt_client_ptr makeClient(::mqtt::connect_options &connOpts, const mq_client_params & params)
	{
		auto target = make_uri(params.host, params.port);
		return makeClient(connOpts, target, params);
	}
	/**
	 * 创建 mqtt::client 实例
	 * @param connOpts [out] connect options
	 * @param uri uri for connect mqtt server,just like 'tcp://127.0.0.1:1883'
	 * @param connect_callback
	 * @return
	 */
	inline mqtt_client_ptr makeClient(::mqtt::connect_options &connOpts, const uri &uri)
	{
		return makeClient(connOpts, mq_client_params(uri));
	}
	/**
	 * 创建 mqtt::client 实例
	 * @param connOpts [out] connect options
	 * @param params parameters for connect mqtt server
	 * @return
	 */
	_BASIC_JSON_TPL_DECLARATION_
	inline mqtt_client_ptr makeClient(::mqtt::connect_options &connOpts, const _BASIC_JSON_TPL_ & params)
	{
		return makeClient(connOpts, mq_client_params(params));
	}

	/**
	 * MQTT 消息订阅实现
	 */
	class MqttSubscriber : public virtual BaseSubscriber, virtual ::mqtt::callback {
	private:
		const int qos;
		mqtt_async_client_ptr client;
		::mqtt::connect_options connOpts;
	protected:
		void doSubscribe(const std::vector<std::string>& channels) override	{
			open();
			std::for_each(channels.begin(), channels.end(), [&](const std::string &c) {
				client->subscribe(c, qos);
			});
		}

		void doUnsubscribe(const std::vector<std::string>& channels) override
		{
			std::for_each(channels.begin(), channels.end(), [&](const std::string c) {
				client->unsubscribe(c);
			});
		}

		void open() override{}

	public:
		/**
		 * 构造函数
		 * @param params parameters for connect mqtt server
		 */
		MqttSubscriber(const mq_client_params& params): qos(2)
		{
			auto target = make_uri(params.host, params.port);
			client = makeClient<::mqtt::async_client>(connOpts, target, params);
			connOpts.set_clean_session(false);
			connOpts.set_automatic_reconnect(true);
			connOpts.set_keep_alive_interval(std::chrono::seconds(20));
			client->set_callback(*this);
			client->connect(connOpts)->wait();
		}
		/** 禁用移动构造函数 */
		MqttSubscriber(MqttSubscriber&&) = delete;
		/** 禁用拷贝构造函数 */
		MqttSubscriber(const MqttSubscriber&) = delete;
		/** 禁用赋值操作符 */
		MqttSubscriber& operator=(const MqttSubscriber&)  = delete;
		virtual ~MqttSubscriber()
		{
			client->disconnect();
		}
		std::string to_string() const
		{
			return client->get_server_uri();
		}
		// connection success
		void connected(const std::string& cause) override {
			SAMPLE_DEBUG_LOG("Connection success");
			/** 连接成功时注册所有频道 */
			for (auto ch : subChannelSet) {
				SAMPLE_DEBUG_LOG("subscribe channel {} using QoS{}",ch,qos);
				client->subscribe(ch, qos);
			}
		}

		// Callback for when the connection is lost.
		void connection_lost(const std::string& cause) override {
			SAMPLE_DEBUG_LOG("Connection lost ");
			if (!cause.empty())
			{
				SAMPLE_DEBUG_LOG("cause: {}", cause);
			}
		}

		// Callback for when a message arrives.
		void message_arrived(::mqtt::const_message_ptr msg) override {
			//SAMPLE_DEBUG_LOG("Message arrived topic: '{}',payload: '{}'", msg->get_topic(), msg->to_string());
			dispatch(msg->get_topic(), msg->to_string());
		}
	};

	///////////////////////////////////////////
	/**
	 * 基于 mqtt 实现的消息发布接口
	 */
	class mqtt_publisher : public publisher {
	public:
		int64_t publish(const std::string &channel, const std::string& message) override
		{		
			if (!cli->is_connected())
			{
				cli->reconnect();
			}
			auto pubmsg = ::mqtt::make_message(channel, message);
			pubmsg->set_qos(qos);
			cli->publish(pubmsg);
			return 0;
		}
		mqtt_publisher(const mqtt_client_ptr &cli) : cli(cli) {
			assert(cli);
		}
		virtual ~mqtt_publisher() = default;
	private:
		int qos = 2;
		mqtt_client_ptr cli;
	};
	/************************************************************************/
	/*            MQTT Client Manager 单实例管理                            */
	/*            必须以全局方式定义一个单实例                              */
	/************************************************************************/
	class ClientManager : public virtual BaseClientManager,public std::enable_shared_from_this<ClientManager> {
	public:
		ClientManager() = default;
		~ClientManager() = default;
		/* 禁用拷贝构造函数 */
		ClientManager(const ClientManager&) = delete;
		/* 禁用赋值操作符 */
		ClientManager& operator=(const ClientManager&) = delete;
		/**
		 * 创建 mqtt::client 实例,如果实例已经存在则直接返回存在的实例
		 * @param params parameters to be connected to
		 * @param connect_callback
		 */
		mqtt_client_ptr getClient(const mq_client_params& params)
		{
			std::lock_guard<std::mutex> guard(clts_lock);
			auto target = params.normalize_uri();
			if (!client_instances.count(target)) {
				::mqtt::connect_options connOpts;
				auto cli = makeClient(connOpts, target, params);
				connOpts.set_keep_alive_interval(std::chrono::seconds(20));
				connOpts.set_clean_session(true);
				cli->connect(connOpts);
				client_instances[target] = cli;
			}
			return client_instances[target];
		}
		/**
 		 * 创建 mqtt::client 实例,如果实例已经存在则直接返回存在的实例
		 * @param params parameters for connect mqtt server
		 */
		_BASIC_JSON_TPL_DECLARATION_
		mqtt_client_ptr getClient(const _BASIC_JSON_TPL_ & params)
		{
			return getClient(mq_client_params(params));
		}
		/**
		 * 创建 mqtt::client 实例,如果实例已经存在则直接返回存在的实例
		 * @param uri uri for connect mqtt server,just like 'tcp://127.0.0.1:1883'
		 * @param connect_callback
		 */
		mqtt_client_ptr getClient(const uri &uri)
		{
			return getClient(mq_client_params(uri));
		}

		/**
		 * 返回host,port指定的mqtt服务的MqttSubscriber实例引用,
		 * @param params parameters to be connected to
		 */
		subscriber_ptr getSubscriber(const mq_client_params& params) {
			std::lock_guard<std::mutex> guard(subs_lock);
			auto target = params.normalize_uri();
			if (!sub_instances.count(target)) {
				/** 不存在则创建新实例 */
				sub_instances[target] = std::make_shared<MqttSubscriber>(params);
			}
			return sub_instances[target];
		}
		/**
		 * 返回host,port指定的mqtt服务的MqttSubscriber实例引用,
		 * @param params parameters to be connected to
		 */
		subscriber_ptr getSubscriber(const base_client_params_ptr& params) override {
			auto ptr = std::dynamic_pointer_cast<mq_client_params>(params);
			throw_if_msg(!ptr, "params is not pointer of mq_client_params");
			return getSubscriber(*ptr);
		}
		/**
		 * 返回访问uri指定的mqtt服务的MqttSubscriber实例引用,
		 * 对于每一个不同的uri,返回对应的单实例
		 * @param uri uri for connect mqtt server,just like 'tcp://127.0.0.1:1883'
		 */
		subscriber_ptr getSubscriber(const uri &uri) override
		{
			return getSubscriber(mq_client_params(uri));
		}
		/**
		 * 返回访问uri指定的mqtt服务的MqttSubscriber实例引用,
		 * 对于每一个不同的uri,返回对应的单实例
		 * @param params parameters for connect mqtt server
		 */
		subscriber_ptr getSubscriber(const simplemq::json &params) override
		{
			return getSubscriber(mq_client_params(params));
		}
		/**
		 * 创建 IPublisher 消息发布实例
		 * @param params parameters for connect mqtt server
		 */
		publisher_ptr getPublisher(const mq_client_params & params) {
			auto cli = getClient(params);
			auto pub = std::make_shared<mqtt_publisher>(cli);
			return std::make_shared<IPublisher>(pub);
		}
		/**
		 * 创建 IPublisher 消息发布实例
		 * @param params parameters for connect mqtt server
		 */
		publisher_ptr getPublisher(const base_client_params_ptr & params) override {
			auto ptr = std::dynamic_pointer_cast<mq_client_params>(params);
			throw_if_msg(!ptr, "params is not pointer of mq_client_params");
			return getPublisher(*ptr);
		}
		/**
	 	 * 创建 IPublisher 消息发布实例
		 * @param uri uri for connect mqtt server,just like 'tcp://127.0.0.1:1883'
		 */
		publisher_ptr getPublisher(const uri & uri)  override {
			return getPublisher(mq_client_params(uri));
		}
		/**
		 * 创建 IPublisher 消息发布实例
		 * @param params parameters for connect mqtt server
		 */
		publisher_ptr getPublisher(const simplemq::json & params) override {
			return getPublisher(mq_client_params(params));
		}
		std::string getImplType()const override
		{
			return "MQTT";
		}
		base_client_params_ptr asClientParam(const simplemq::json & params) override {
			return std::make_shared<mq_client_params>(params);			
		}
	private:
		winsock_initializer winsock_init;
		std::map<uri, mqtt_client_ptr> client_instances;
		std::mutex clts_lock;
		std::map<uri, subscriber_ptr> sub_instances;
		std::mutex subs_lock;
	};	
} /** namespace simplemq */
#endif /* SIMPLEMQ_MQTT_SIMPLEMQ_MQTT_HPP_ */
